package com.shujia.flink.core;

import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

import java.time.Duration;

public class Demo05EventTime {
    public static void main(String[] args) throws Exception {
        // 事件时间：数据自带的时间(基于毫秒的时间戳)，Flink可以使用数据中的时间来确定是否触发窗口
        // 每5s统计一次，基于事件时间的滚动窗口

        /*
        准备数据：
        1744447972000 2025-04-12 16:52:52

        a,1744447972000
        a,1744447973000
        a,1744447974000
        a,1744447975000
        a,1744447976000
        a,1744447977000
        a,1744447978000
        a,1744447979000
        a,1744447981000
        a,1744447985000
         */

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 将并行度设置为1更好的观察结果
        // 思考：如果并行度大于1，水位线如何计算？窗口如何触发？
        env.setParallelism(1);

        DataStreamSource<String> socketDS = env.socketTextStream("master", 8888);

        // 解析数据
        SingleOutputStreamOperator<Tuple2<String, Long>> kvDS = socketDS.map(line -> {
            String[] splits = line.split(",");
            String word = splits[0];
            long ts = Long.parseLong(splits[1]);
            return Tuple2.of(word, ts);
        }, Types.TUPLE(Types.STRING, Types.LONG));

        // 设置时间戳以及水位线的策略
        SingleOutputStreamOperator<Tuple2<String, Long>> assignDS = kvDS.assignTimestampsAndWatermarks(
                /*
                 * 水位线：根据一定的水位线策略基于数据中的事件时间计算而来
                 * 水位线策略：
                 *  1、单调递增的水位线策略：取当前线程中所接收到的数据最大的时间作为水位线
                 *  2、固定时间的乱序容忍水位线策略（水位线前移策略）：假设固定时间为5s，水位线=当前线程中所接收到的数据最大的时间-固定时间5s
                 */
                WatermarkStrategy
                        // 每个线程会单独维护一个水位线
//                        .<Tuple2<String, Long>>forMonotonousTimestamps() // 连续单调递增的水位线策略，取接收到的数据中最大的时间作为水位线
                        // 水位线前移策略：将水位线减固定的时间
                        .<Tuple2<String,Long>>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                        // 告诉Flink数据中的哪一部分是事件时间
                        .withTimestampAssigner(new SerializableTimestampAssigner<Tuple2<String, Long>>() {
                            @Override
                            public long extractTimestamp(Tuple2<String, Long> kv, long l) {
                                return kv.f1;
                            }
                        })
        );

        assignDS
                // 将数据变成 word,1的KV格式
                .map(kv -> Tuple2.of(kv.f0, 1), Types.TUPLE(Types.STRING, Types.INT))
                .keyBy(t2 -> t2.f0)
                /*
                窗口的触发条件：水位线超过了窗口的结束时间
                 */
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .sum(1)
                .print();


        env.execute();

    }
}
